package com.gmall.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gmall.app.function.DimAsyncFunction;
import com.gmall.bean.OrderGoods;
import com.gmall.bean.OrderInfo;
import com.gmall.bean.OrderWide;
import com.gmall.utils.MykafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.concurrent.TimeUnit;

public class OrderWideApp {
    public static void main(String[] args) throws Exception {
        //获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //读取kafka主题的数据，并转换为JavaBean对象提取时间戳生成watermark
        String orderInfoSourceTopic = "dwd_order_info";
        //todo 迁移库数据不完整可能要改字段
        String orderGoodsSourceTopic = "dwd_order_goods";
        // 订单宽表topic
        String orderWideSinkTopic = "dwd_order_wide";
        String groupId = "order_wide_group_2022";

        SingleOutputStreamOperator<OrderInfo> orderInfoDS = env.addSource(MykafkaUtil.getKafkaConsumer(orderInfoSourceTopic, groupId)).map(e -> {
            OrderInfo orderInfo = JSON.parseObject(e, OrderInfo.class);
            String create_time = orderInfo.getCreate_time();
            String[] dateTimeArr = create_time.split(" ");
            orderInfo.setCreate_date(dateTimeArr[0]);
            orderInfo.setCreate_hour(dateTimeArr[1].split(":")[0]);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            orderInfo.setCreate_ts(sdf.parse(create_time).getTime());

            return orderInfo;
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                    @Override
                    public long extractTimestamp(OrderInfo orderInfo, long l) {
                        return orderInfo.getCreate_ts();
                    }
                }));

        SingleOutputStreamOperator<OrderGoods> orderGoodsDS = env.addSource(MykafkaUtil.getKafkaConsumer(orderGoodsSourceTopic, groupId)).map(e -> {
            OrderGoods orderGoods = JSON.parseObject(e, OrderGoods.class);
            String create_time = orderGoods.getCreate_time();
            String[] datatimeArr = create_time.split(" ");
            orderGoods.setCreate_date(datatimeArr[0]);
            orderGoods.setCreate_hour(datatimeArr[1].split(":")[0]);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            orderGoods.setCreate_ts(sdf.parse(create_time).getTime());

            return orderGoods;
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderGoods>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderGoods>() {
                    @Override
                    public long extractTimestamp(OrderGoods orderGoods, long l) {
                        return orderGoods.getCreate_ts();
                    }
                }));

        //TODO 双流join
        SingleOutputStreamOperator<OrderWide> process = orderInfoDS.keyBy(OrderInfo::getOrder_id)
                .intervalJoin(orderGoodsDS.keyBy(OrderGoods::getOrder_id))
                .between(Time.seconds(-5), Time.seconds(5)) //给最大延迟时间
                .process(new ProcessJoinFunction<OrderInfo, OrderGoods, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo orderInfo, OrderGoods orderGoods, Context context, Collector<OrderWide> collector) throws Exception {
                        collector.collect(new OrderWide(orderInfo, orderGoods));
                    }
                });

        //打印测试
        process.print("orderWideWithNoDimDS>>>>>");

        //关联用户维度
        AsyncDataStream.unorderedWait(
                process,
                new DimAsyncFunction<OrderWide>("DIM_USER_INFO"){

                    @Override
                    public String getKey(OrderWide input) {
                        return input.getUser_id().toString();
                    }

                    @Override
                    public void join(OrderWide input, JSONObject dimInfo) throws ParseException {
                        input.setUser_name(dimInfo.getString("NAME"));
                        String birthday = dimInfo.getString("BIRTHDAY");
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                        long currentTimeMillis = System.currentTimeMillis();
                        long time = sdf.parse(birthday).getTime();
                        long age = (currentTimeMillis-time)/(1000*60*60*24*365L);
                        input.setUser_age((int)age);
                    }
                },60, TimeUnit.SECONDS
        );

        AsyncDataStream.unorderedWait(
                process,
                new DimAsyncFunction<OrderWide>("DIM_SKU_INFO"){

                    @Override
                    public String getKey(OrderWide input) {
                        return null;
                    }

                    @Override
                    public void join(OrderWide input, JSONObject dimInfo) throws ParseException {

                    }
                },60,TimeUnit.SECONDS
        );

        env.execute();
    }


}
